# -*- coding: utf-8 -*-
# @Time        :2024/11/2 下午3:32
# @Author      :文刀水寿
# @File        : 04_数据计算_案例2.py
"""
 @Description :
"""
from pyspark import SparkConf, SparkContext
import os
import json

os.environ['PYSPARK_PYTHON'] = "D:/Python/python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)

# T0D0  需求1： 城市销售额排名
# 1.1 读取文件
file_rdd = sc.textFile("D:/orders.txt")

# 1.2 取出一个个json字符串
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))

# 1.3将一个个json字符串转换为字典
dict_str_rdd = json_str_rdd.map(lambda x: json.loads(x))

# 1.4 取出城市和销售额数据
city_money = dict_str_rdd.map(lambda x: (x['areaName'], int(x['money'])))

# 1.5 按城市分组按销售额聚合
result_rdd = city_money.reduceByKey(lambda a, b: a + b)

# 1.6 按销售额聚合结果进行排序
result1 = result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
print("需求1 的结果是：", result1.collect())

# 2.1 取出全部的商品类别
# 2.2 对全部商品类别进行去重
category_rdd = dict_str_rdd.map(lambda x: (x["category"])).distinct()
print("需求2 的类别有：", category_rdd.collect())

# 3.1 过滤北京市的数据
beijing_rdd = dict_str_rdd.filter(lambda x: x["areaName"] == "北京")

# 3.2 进行商品类别去重
result3 = beijing_rdd.map(lambda x: x['category']).distinct()
print("需求3 的结果是：", result3.collect())
